/**
 * @Project Name :  als-rpc
 * @Package Name :  com.alvis.registry.zookeeper
 * @Description :  TODO
 * @author :  Alvis
 * @Creation Date:  2018-07-06 9:24 AM
 * @ModificationHistory Who    When    What
 * --------  ---------  --------------------------
 */
package com.alvis.registry.zookeeper;

import com.alvis.utilities.Component;
import com.alvis.utilities.NetWorkHelper;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author :  Alvis
 * @Description :  TODO
 * @Creation Date:  2018-07-06 9:24 AM
 */

public class ServiceNode implements Component {

    private final static Logger logger = LoggerFactory.getLogger(ServiceNode.class);
    private final static String rootNode = "/alsrpc";
    private final static String node_format = rootNode + "/{service}/providers";
    private final static String providerPathTag = "/providers/";

    private CuratorFramework curatorFramework;
    private TreeCache treeCache;
    private ConcurrentHashMap<String, Vector<String>> zkServiceNames = new ConcurrentHashMap<>();


    public ServiceNode(CuratorFramework curatorFramework) {
        this.curatorFramework = curatorFramework;
    }


    @Override
    public void start() {
        loadService();
        listener();
    }

    @Override
    public void stop() {
        treeCache.close();
    }


    public void addNode(String serviceName) {
        try {
            String node = node_format.replace("{service}", serviceName) + "/" + NetWorkHelper.getLocalHost();
            curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(node);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public String getRandomService(String serviceName) {
        Vector<String> urls = zkServiceNames.get(serviceName);
        java.util.Random random = new java.util.Random();
        return urls.get(random.nextInt(urls.size()));
    }


    private void loadService() {
        try {
            GetChildrenBuilder childrenBuilder = curatorFramework.getChildren();
            childrenBuilder.forPath(rootNode).forEach(s -> {
                try {
                    List<String> ipNodes = childrenBuilder.forPath(node_format.replace("{service}", s));
                    if (0 != ipNodes.size()) {
                        Vector<String> ips = new Vector<>(ipNodes.size());
                        ipNodes.forEach(ip -> {
                            ips.add(ip);
                        });
                        zkServiceNames.put(s, ips);
                    }
                } catch (Exception e) {
                    logger.error(e.toString());
                }
            });
        } catch (Exception e) {
            logger.error(e.toString());
        }
    }


    private void listener() {
        treeCache = new TreeCache(curatorFramework, rootNode);
        treeCache.getListenable().addListener((client, event) -> {
            ChildData data = event.getData();
            if (data != null) {
                String ipPath = data.getPath();
                if (ipPath.contains(providerPathTag)) {
                    String[] paths = ipPath.split("/");
                    switch (event.getType()) {
                        case NODE_ADDED:
                            if (zkServiceNames.containsKey(paths[2])) {
                                Vector<String> ips = zkServiceNames.get(paths[2]);
                                if (ips.contains(paths[4])) {
                                } else {
                                    ips.add(paths[4]);
                                }
                            } else {
                                Vector<String> ips = new Vector<>();
                                ips.add(paths[4]);
                                zkServiceNames.put(paths[2], ips);
                            }
                            logger.info("NODE_ADDED : " + ipPath + "  Data:" + new String(data.getData()));
                            break;
                        case NODE_REMOVED:
                            if (zkServiceNames.containsKey(paths[2])) {
                                Vector<String> ips = zkServiceNames.get(paths[2]);
                                if (ips.contains(paths[4])) {
                                    ips.remove(paths[4]);
                                } else {

                                }
                            } else {

                            }
                            logger.info("NODE_REMOVED : " + ipPath + "  Data:" + new String(data.getData()));
                            break;
                        case NODE_UPDATED:
                            logger.info("NODE_UPDATED : " + ipPath + "  Data:" + new String(data.getData()));
                            break;
                        default:
                            break;
                    }
                }
            } else {
            }
        });
        try {
            treeCache.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
